#pragma once

#include "dce.h"
#include "Log.h"
#include <vector>
using std::vector;

class pure_message : public message
{
	buffer _body_buf;
public:
	void set_buffer(const buffer &buf)
	{
		_body_buf = buf;
	}
	int marshal(buffer &buf)
	{
		octet_ostream oos;
		set_length(_body_buf.get_size());
		oos << get_head();
		oos.append_data(_body_buf.get_data_buf(), _body_buf.get_size());
		buf = oos.get_buffer();
		return 0;
	}
	int unmarshal(buffer &buf)
	{
		_body_buf = buf;
		return 0;
	}
	int response(worker *wkr);
};

class dispatch_server;
class dispatch_listener_handler : public read_event_handler
{
	dispatch_server *_svr;
public:
	int set_server(dispatch_server *svr) { _svr = svr; }
	int read(socket_event *se);
};

class dispatch_client_handler : public read_event_handler
{
	dispatch_server *_svr;
public:
	dispatch_client_handler(dispatch_server *svr) : _svr(svr) {};
	int read(socket_event *se);
};

class dispatch_server : public server, private pthread
{
	thread_pool _msg_thread_pool;
	vector<address> _front_servers;
	vector<int> _front_servers_weight;
	address _addr;
public:
	~dispatch_server() {
		LOG_INFO_VA("dispatch server destroy");
	}
	void set_listener_address(const address &addr) { _addr = addr; }
	thread_pool *get_thread_pool() { return &_msg_thread_pool; }
	void set_msg_worker(worker *msgwkr)
	{
		_msg_thread_pool.set_worker(msgwkr);
	}
	
	int thread_func()
	{
		int svr_sock = socket(PF_INET, SOCK_STREAM, 0);
		if (svr_sock < 0) {
			LOG_ERROR_VA("[%s] create server socket failed", this->whoami().c_str());
			return -1;
		}
		int flag = 1;
		setsockopt(svr_sock, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(int));
		if (bind(svr_sock, &_addr, _addr.length()) < 0) {
			LOG_ERROR_VA("[%s] bind address failed", this->whoami().c_str());
			return -1;
		}
		
		int sockflag = fcntl(svr_sock, F_GETFL, 0);
		fcntl(svr_sock, F_SETFL, sockflag | O_NONBLOCK);
		
		if (listen(svr_sock, 100) < 0) {
			LOG_ERROR_VA("[%s] listen failed", this->whoami().c_str());
			return -1;
		}
		socket_event *listener = new socket_event(socket_event::read);
		listener->set_handle(svr_sock);
		dispatch_listener_handler *eh = new dispatch_listener_handler();
		eh->set_server(this);
		listener->set_event_handler(eh);
		set_listener_event(listener);
		server::startup();
	}
	int startup()
	{
		return pthread::start();
	}
	int waitend()
	{
		return pthread::wait();
	}
	int stop()
	{
		return pthread::kill();
	}
	void add_front_node(const address &addr, int weight) { 
		_front_servers.push_back(addr); 
		_front_servers_weight.push_back(weight); 
	}
	int setup();
};

class dispatch_worker : public worker
{
	server_proxy_duplication _front_nodes;

public:
	void add_front_node(const address &addr, int weight) { _front_nodes.add_server_proxy(server_proxy(addr), weight); }
	server_proxy *choose_one_front_node() { return _front_nodes.choose_one(); }
	worker *clone() { return new dispatch_worker(*this); }
};


